package com.qf.config;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.qf.entity.Message;
import com.qf.mapper.MessageMapper;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.List;

/**
 * 作者：大狗
 * 时间：2021/7/13 14:22
 */
@Configuration
public class MessageCompensate {
    private static final String PAY_NAME = "pay_queue";
    private static final String PAY_EXCHANGE = "pay_exchange";
    private static final String PAY_KEY = "pay_key";

    private MessageMapper messageMapper;
    private RabbitTemplate availableTemplate;

    public MessageCompensate(MessageMapper messageMapper, RabbitTemplate availableTemplate) {
        this.messageMapper = messageMapper;
        this.availableTemplate = availableTemplate;
    }

    //消息补偿, 就是在数据的 message 表中查询重试次数小于 5 的消息
    @Scheduled(cron = "0/10 * * ? * *")
    public void reTrySendMessage(){
        QueryWrapper queryWrapper = new QueryWrapper();
        queryWrapper.lt("try_count", 5);
        queryWrapper.gt("try_count", 0);
        List<Message> list = messageMapper.selectList(queryWrapper);
        list.stream().forEach(m ->{
            CorrelationData cd = new CorrelationData(m.getId());
            MessageProperties mp = new MessageProperties();
            mp.setCorrelationId(m.getId());
            mp.setDeliveryMode(MessageDeliveryMode.PERSISTENT);  // 表示消息为持久化消息
            //设置消息多少秒达到死信交换机 15s
            mp.setExpiration("1500");
            org.springframework.amqp.core.Message msg = new org.springframework.amqp.core.Message(m.getContent().getBytes(), mp);
            // 重新投递
            availableTemplate.convertAndSend(m.getExchange(), m.getRoutingKey(), m.getContent(), cd);
        });
    }
}
